package edu.ustb.efws.pipeline.collect.handler.meter;

import edu.ustb.efws.EfwsProperties;
import edu.ustb.efws.pipeline.domain.RiskLog;
import edu.ustb.efws.meter.service.MeterService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * 测量点Pool
 *
 * @author TS Group
 */
public class MeterPointPool {
    private static final Logger LOGGER = LoggerFactory.getLogger(MeterPointPool.class);

    private static final long WAIT_UPDATE_MILLS = Duration.ofMinutes(60).toMillis();

    private final long windowTimeMill;
    private final ConcurrentMap<Integer, ConcurrentMap<Integer, MeterPoint>> pool;
    private final MeterService service;

    /**
     * 构造{@link MeterPointPool}
     *
     * @param properties {@link EfwsProperties}
     * @param service {@link MeterService}
     */
    public MeterPointPool(EfwsProperties properties, MeterService service) {
        this.windowTimeMill = properties.getWindowTimeMin() * 60 * 1000L;
        this.service = service;
        this.pool = new ConcurrentHashMap<>();
    }

    /**
     * 添加测量值
     *
     * @param log {@link RiskLog}
     */
    public void push(RiskLog log){

        ConcurrentMap<Integer, MeterPoint> bucket = pool.get(log.getPositionId());
        if(bucket == null){
            bucket = new ConcurrentHashMap<>(20);
            ConcurrentMap<Integer, MeterPoint> oBucket = pool.putIfAbsent(log.getPositionId(), bucket);
            if(oBucket != null){
                bucket = oBucket;
            }
        }

        int windowTimeIndex = windowTimeIndex(log.getIptTime().getTime());
        MeterPoint point = bucket.get(windowTimeIndex);
        if(point == null){
            point = new MeterPoint(log.getPositionId(), windowTimeIndex, log.getScore());
            MeterPoint oPoint = bucket.putIfAbsent(windowTimeIndex, point);
            if(oPoint != null){
                oPoint.incScore(log.getScore());
            }
        }else{
            point.incScore(log.getScore());
        }
    }

    private int windowTimeIndex(long collectTime){
        return (int)(collectTime/windowTimeMill);
    }

    /**
     * 保存修改的测量点信息
     */
    public void saveUpdate(){

        long dieOutTime = System.currentTimeMillis() - WAIT_UPDATE_MILLS;

        List<MeterPoint> removePoints = new ArrayList<>();
        for(Map<Integer, MeterPoint> bucket: pool.values()){
            for(Map.Entry<Integer, MeterPoint> entry: bucket.entrySet()){
                MeterPoint point = entry.getValue();
                if(point.getUpdateTime() > point.getRepoTime()){
                    if(save(point)){
                        point.setRepoTime(System.currentTimeMillis());
                    }
                }else if(point.getUpdateTime() < dieOutTime){
                   removePoints.add(point);
                }
            }
        }

        for(MeterPoint point: removePoints){
            boolean ok = pool.get(point.getPositionId()).remove(point.getIndex()) != null;
            LOGGER.info("Remove meter value pointId={}, windowTimeIndex={}, success={}",
                    point.getPositionId(), point.getIndex(), ok);

            if(point.getUpdateTime() > point.getRepoTime()){
                save(point);
            }
        }
    }

    private boolean save(MeterPoint point){
        try{
            if(!service.update(point)){
                service.save(point);
            }
            return true;
        }catch (Exception ex){
            LOGGER.error("Save meter fail point={}, error={}", point.toString(), ex.getMessage());
            return false;
        }
    }

    /**
     * 保存所有测量点信息
     */
    public void saveAll(){
        for(Map.Entry<Integer, ConcurrentMap<Integer, MeterPoint>> bucket: pool.entrySet()){
            LOGGER.info("Save all meter positionId ={}, count={}", bucket.getKey(), bucket.getValue().size());
            for(MeterPoint point: bucket.getValue().values()) {
                if(point.getUpdateTime() > point.getRepoTime()){
                    save(point);
                }
            }
        }
    }
}
